package rpcmon

import (
	"encoding/json"
	"errors"
	"net/http"
	"net/url"
	"strings"
	"sync"
	"time"

	"gitee.com/JMArch/rpcmon-gateway/conf"
	"gitee.com/JMArch/rpcmon-gateway/log"
	"gitee.com/JMArch/rpcmon-gateway/storage"
	"gitee.com/JMArch/rpcmon-gateway/util"
	"github.com/gorilla/websocket"
	"go.uber.org/zap"
)

type ClientGroup struct {
	rw    sync.RWMutex
	group map[string]*Client
}

type Client struct {
	log      *zap.Logger
	url      *url.URL
	wg       sync.WaitGroup
	conn     *websocket.Conn
	cache    *storage.Cache
	lifeTime time.Duration
	interval time.Duration
	dialer   *websocket.Dialer
	name     string
	quit     chan bool
	mu       sync.RWMutex
	retry    uint64
	services map[string]bool
}

func NewClientGroup(cacheGroup *storage.CacheGroup, config *conf.GatewayConf, logger *log.GatewayLog) (*ClientGroup, error) {
	g := &ClientGroup{
		group: make(map[string]*Client, 0),
	}

	for name, cache := range cacheGroup.RPCGroup {
		c, err := New(cache, config, logger)
		if err != nil {
			return nil, err
		}
		g.group[name] = c
	}

	return g, nil
}

func (g *ClientGroup) Close() {
	for _, client := range g.group {
		client.Close()
	}
	return
}

func New(cache *storage.Cache, config *conf.GatewayConf, logger *log.GatewayLog) (*Client, error) {
	name := cache.Name

	b, ok := config.Rpcmon.BackendMap[name]
	if !ok {
		return nil, errors.New("backend name not exists")
	}

	if b.History > config.Rpcmon.MaxHistory {
		b.History = config.Rpcmon.MaxHistory
	}

	client := &Client{
		name:     name,
		log:      logger.Named("rpcmon." + name),
		cache:    cache,
		lifeTime: b.History,
		interval: b.Interval,
		quit:     make(chan bool, 0),
	}

	m, err := util.GetRPCServicesList(config.Rpcmon.EtcdAddr)
	if err != nil {
		return nil, err
	}
	client.services = m

	u := &url.URL{
		Scheme: "ws",
		Host:   b.Addr,
		Path:   b.Path,
	}
	client.url = u

	d := &websocket.Dialer{
		Proxy:            http.ProxyFromEnvironment,
		HandshakeTimeout: 10 * time.Second,
	}
	client.dialer = d

	conn, _, err := d.Dial(u.String(), nil)
	if err != nil {
		return nil, err
	}
	client.conn = conn

	if name == "errlog" {
		conn.WriteMessage(websocket.TextMessage, []byte("{\"attr\":\"set:json\"}"))
	} else {
		conn.WriteMessage(websocket.TextMessage, []byte("set:json"))
	}

	client.WriteLoop()
	client.ReadLoop()

	return client, nil
}

func (c *Client) WriteLoop() {
	c.wg.Add(1)
	go func() {

		defer func() {
			if p := recover(); p != nil {
				c.log.Error("write panic", zap.Any("error", p))
			}
			c.wg.Done()
		}()

		pingTicker := time.NewTicker(1 * time.Second)
		defer pingTicker.Stop()
		for {
			select {
			case <-c.quit:
				return
			case <-pingTicker.C:

				pingData := []byte{}
				if c.name == "errlog" {
					pingData = []byte("{\"attr\":\"ping_data\"}")
				}
				err := c.GetConn().WriteMessage(websocket.PingMessage, pingData)
				if err != nil {
					c.log.Error("websocket retry", zap.String("type", c.name), zap.String("local", c.GetConn().LocalAddr().String()), zap.Error(err))
					if retryErr := c.Retry(err); retryErr != nil {
						return
					}
				}
			}
		}
	}()

}

func (c *Client) GetConn() *websocket.Conn {
	c.mu.RLock()
	defer c.mu.RUnlock()

	return c.conn
}

func (c *Client) Retry(connErr error) error {

	defer func() {
		if p := recover(); p != nil {
			c.log.Error("retry panic", zap.Any("error", p))
		}
	}()

	select {
	case <-c.quit:
		c.log.Error("websocket closed", zap.String("type", c.name))
		return errors.New("client closed")
	default:
	}

	c.mu.Lock()
	defer c.mu.Unlock()

	old := c.conn.LocalAddr().String()

	//if e, ok := connErr.(*websocket.CloseError); ok {
	//	if e.Code != websocket.CloseAbnormalClosure {
	//		c.log.Error("websocket reconnect refused", zap.String("type", c.name), zap.String("local", old), zap.Error(connErr))
	//		return connErr
	//	}
	//} else {
	//	c.log.Error("websocket reconnect refused", zap.String("type", c.name), zap.String("local", old), zap.Error(connErr))
	//	return connErr
	//}

	c.retry += 1

	if err := c.conn.Close(); err != nil {
		c.log.Warn("websocket close conn fail", zap.String("type", c.name), zap.String("local", old), zap.Error(err))
	}

	time.Sleep(5 * time.Second)

	var retryErr error
	c.conn, _, retryErr = c.dialer.Dial(c.url.String(), nil)
	if retryErr != nil {
		c.log.Error("websocket reconnect fail", zap.String("type", c.name), zap.String("local", old), zap.Uint64("retry", c.retry), zap.Error(retryErr))
		return retryErr
	}

	if c.name == "errlog" {
		c.conn.WriteMessage(websocket.TextMessage, []byte("{\"attr\":\"set:json\"}"))
	} else {
		c.conn.WriteMessage(websocket.TextMessage, []byte("set:json"))
	}

	c.log.Warn("websocket reconnect succ", zap.String("type", c.name), zap.String("old", old), zap.String("new", c.conn.LocalAddr().String()), zap.Uint64("retry", c.retry), zap.Error(connErr))
	return nil
}

func (c *Client) getServiceName(k string) string {
	if k == "Optool-service" {
		k = "optool-service"
	} else if k == "MobileService" {
		k = "Mobile"
	} else if k == "product-service" {
		k = "ProductService"
	} else if k == "operation-service" {
		k = "Operationservice"
	} else if k == "TrusteeshipData" {
		k = "Trusteeship"
	} else if k == "openapi-service" {
		k = "openapi"
	} else if k == "ERPService" {
		k = "erp-service"
	} else if k == "Koubei" {
		k = "koubei-service"
	} else if k == "meidian_crm_service" {
	} else if k == "cart" {
	}

	if _, ok := c.services[k]; ok {
		return k
	}

	return ""

}

func (c *Client) ReadLoop() {
	c.wg.Add(1)
	go func() {

		defer func() {
			if p := recover(); p != nil {
				c.log.Error("read panic", zap.Any("error", p))
			}
			c.wg.Done()
		}()

		for {
			mType, message, err := c.GetConn().ReadMessage()
			if err != nil {
				c.log.Error("websocket retry", zap.String("type", c.name), zap.String("local", c.GetConn().LocalAddr().String()), zap.Error(err))
				if retryErr := c.Retry(err); retryErr == nil {
					continue
				}
				return
			}

			if string(message) == "[]" {
				continue
			}

			if mType != websocket.TextMessage {
				c.log.Warn("get backend message unexpect", zap.String("type", c.name), zap.Any("messageType", mType), zap.String("message", string(message)))
				continue
			}

			fetch := time.Now()

			if c.name == "qps" {
				hashMap := make(map[string]interface{}, 0)
				err = json.Unmarshal(message, &hashMap)
				if err != nil {
					c.log.Error("invalid message format", zap.Error(err))
				} else {
					if len(hashMap) > 0 {

						allNodes := make([]storage.Node, 0)
						for k, v := range hashMap {
							if newk := c.getServiceName(k); newk != "" {
								node := make(storage.Node, 0)
								node["Service"] = newk
								node["QPS"] = int(v.(float64))
								node["Create"] = fetch
								allNodes = append(allNodes, node)
							} else {
								node := make(storage.Node, 0)
								node["Service"] = k
								node["QPS"] = int(v.(float64))
								node["Create"] = fetch
								allNodes = append(allNodes, node)
								//c.log.Warn("dirty message", zap.String("type", c.name), zap.String("service", k))
							}
						}

						//c.log.Debug("get backend message", zap.String("type", c.name), zap.Int("count", len(allNodes)))
						if err := c.cache.SetAll(allNodes, time.Now(), c.lifeTime, c.interval); err != nil {
							c.log.Error("cache set fail", zap.String("type", c.name), zap.Error(err))
							return
						}
					}
				}
			} else if c.name == "failqps" {
				hashMap := make(map[string]interface{}, 0)
				err = json.Unmarshal(message, &hashMap)
				if err != nil {
					c.log.Error("invalid message format", zap.Error(err), zap.String("data", string(message)))
				} else {
					if len(hashMap) > 0 {

						allNodes := make([]storage.Node, 0)
						for k, v := range hashMap {
							if newk := c.getServiceName(k); newk != "" {
								node := make(storage.Node, 0)
								node["Service"] = newk
								node["Err"] = v
								node["Create"] = fetch
								allNodes = append(allNodes, node)
							} else {
								node := make(storage.Node, 0)
								node["Service"] = k
								node["Err"] = v
								node["Create"] = fetch
								allNodes = append(allNodes, node)
								//c.log.Warn("dirty message", zap.String("type", c.name), zap.String("service", k))
							}
						}

						//c.log.Debug("get backend message", zap.String("type", c.name), zap.Int("count", len(allNodes)))
						if err := c.cache.SetAll(allNodes, time.Now(), c.lifeTime, c.interval); err != nil {
							c.log.Error("cache set fail", zap.String("type", c.name), zap.Error(err))
							return
						}
					}
				}
			} else if c.name == "errlog" {
				arr := make([]string, 0)
				hashMap := make(map[string][]string, 0)
				err = json.Unmarshal(message, &arr)
				if err != nil {
					c.log.Error("invalid message format", zap.Error(err), zap.String("data", string(message)))
				} else {
					for _, info := range arr {
						m := strings.Split(info, "\t")
						if len(m) >= 2 {
							fetch, err = time.Parse("2006-01-02 15:04:05", m[0])
							if err == nil {
								n := strings.Split(m[1], "::")
								if len(n) >= 1 {
									service := n[0]
									if _, ok := hashMap[service]; !ok {
										hashMap[service] = make([]string, 0)
									}
									hashMap[service] = append(hashMap[service], info)
								}
							}
						}
					}

					if len(hashMap) > 0 {

						allNodes := make([]storage.Node, 0)
						for k, v := range hashMap {
							if newk := c.getServiceName(k); newk != "" {
								node := make(storage.Node, 0)
								node["Service"] = newk
								node["Err"] = v
								node["Create"] = fetch
								allNodes = append(allNodes, node)
							} else {
								node := make(storage.Node, 0)
								node["Service"] = k
								node["Err"] = v
								node["Create"] = fetch
								allNodes = append(allNodes, node)
								//c.log.Warn("dirty message", zap.String("type", c.name), zap.String("service", k))
							}
						}

						//c.log.Debug("get backend message", zap.String("type", c.name), zap.Int("count", len(allNodes)))
						if err := c.cache.SetAll(allNodes, time.Now(), c.lifeTime, c.interval); err != nil {
							c.log.Error("cache set fail", zap.String("type", c.name), zap.Error(err))
							return
						}
					}
				}

			}
		}
	}()

	return
}

func (c *Client) Close() {
	close(c.quit)
	c.GetConn().Close()
	c.log.Debug("stop backend, close conn", zap.String("type", c.name))
	c.wg.Wait()
	c.log.Debug("stop backend, wait group", zap.String("type", c.name))
}
